/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.jssdb.core.net;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import org.jssdb.core.event.CustomEvent;
import org.jssdb.core.event.CustomEventListener;
import org.jssdb.core.query.Request;

/**
 *
 * @author developer
 */
public class ConnectorServer {

    private DataChunkCollector dataChunkCollector = DataChunkCollector.getInstance();
    private String remoteHost = "localhost", myHost = "localhost";
    private int remotePort = 5500, myPort = 5501;
    private CarrierCollector carrierCollector = CarrierCollector.getInstance();
    private static final Logger logger = Logger.getLogger(ConnectorServer.class.getName());

    private ConnectorServer() {
        dataChunkCollector.addCustomEventListener(new CustomEventListener() {

            @Override
            public void customEventOccurred(CustomEvent evt) {
                if (evt.getSource() instanceof CarrierRequestResponse) {
                    CarrierRequestResponse crr = (CarrierRequestResponse) evt.getSource();
                    if (crr != null) {
                        carrierCollector.add(crr);
                    }
                }
            }
        });
    }

    public String getMyHost() {
        return myHost;
    }

    public void setMyHost(String myHost) {
        this.myHost = myHost;
    }

    public int getMyPort() {
        return myPort;
    }

    public void setMyPort(int myPort) {
        this.myPort = myPort;
    }

    public String getRemoteHost() {
        return remoteHost;
    }

    public void setRemoteHost(String remoteHost) {
        this.remoteHost = remoteHost;
    }

    public int getRemotePort() {
        return remotePort;
    }

    public void setRemotePort(int remotePort) {
        this.remotePort = remotePort;
    }

    public void start() {
        Executors.newSingleThreadExecutor().execute(reciever());
    }

    public CarrierRequestResponse getResult(CarrierRequestResponse crr, int timeout) throws Exception {
        CarrierRequestResponse result = null;

        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                result = carrierCollector.getCarrierRequestResponse(crr.getId());
                tc.setHasMore(false);
                timer.cancel();
            }
            Thread.sleep(10);
        }

        return result;
    }

    public ArrayList<Map<Long, Object>> getResultSet(CarrierRequestResponse crr, int timeout) throws Exception {
        ArrayList<Map<Long, Object>> result = null;

        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                CarrierRequestResponse obj = carrierCollector.getCarrierRequestResponse(crr.getId());
                if(obj != null){
                    if(obj.getResultObject() instanceof ArrayList){
                        result = (ArrayList<Map<Long, Object>>)obj.getResultObject();
                    }
                }
                tc.setHasMore(false);
                timer.cancel();
            }
            Thread.sleep(10);
        }

        return result;
    }

    public ArrayList<Map<Long, Object>> getResultSet(Request request, int timeout) throws Exception {
        ArrayList<Map<Long, Object>> result = null;
        
        if(request == null || timeout <= 0){
            return result;
        }

        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        CarrierRequestResponse crr = new CarrierRequestResponse(new Date().getTime(),
                getRemoteHost(), getRemotePort(),
                getMyHost(), getMyPort());
        crr.setRequest(request);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                CarrierRequestResponse obj = carrierCollector.getCarrierRequestResponse(crr.getId());
                if(obj != null){
                    if(obj.getResultObject() instanceof ArrayList){
                        result = (ArrayList<Map<Long, Object>>)obj.getResultObject();
                    }
                }
                tc.setHasMore(false);
                timer.cancel();
            }
            Thread.sleep(10);
        }

        return result;
    }

    /**
     * 
     * @param obj
     * @param timeout
     * @return -1 undefined (may be or may be not), 3 - ok, 6 - false
     */
    public long addObject(Object obj, int timeout) {
        long id = -1;
        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        CarrierRequestResponse crr = new CarrierRequestResponse(new Date().getTime(),
                remoteHost, remotePort, myHost, myPort);
        crr.setAction(CarrierRequestResponse.ADD);
        crr.setTransferObject(obj);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                CarrierRequestResponse crrGet = carrierCollector.getCarrierRequestResponse(crr.getId());
                if (crrGet != null) {
                    Object o = crrGet.getResultObject();
                    if (o != null && o instanceof Long) {
                        id = (Long) o;
                    }
                }

                tc.setHasMore(false);
                timer.cancel();
            }
        }

        return id;
    }

    public int updateObject(long id, Object obj, int timeout) {
        int result = -1;
        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        CarrierRequestResponse crr = new CarrierRequestResponse(new Date().getTime(),
                remoteHost, remotePort, myHost, myPort);
        crr.setAction(CarrierRequestResponse.MODIFY);
        crr.setTransferObjectId(id);
        crr.setTransferObject(obj);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                result = carrierCollector.getCarrierRequestResponse(crr.getId()).getStatus();
                tc.setHasMore(false);
                timer.cancel();
            }
        }

        return result;
    }

    public int deleteObject(long id, String className, int timeout) {
        int result = -1;
        TimeoutConnector tc = new TimeoutConnector();
        Timer timer = new Timer();
        Date date = new Date();
        date.setTime(date.getTime() + timeout);
        timer.schedule(tc, date);

        CarrierRequestResponse crr = new CarrierRequestResponse(new Date().getTime(),
                remoteHost, remotePort, myHost, myPort);
        crr.setAction(CarrierRequestResponse.DELETE);
        crr.setTransferObjectId(id);
        crr.setTransferObject(className);

        send(crr);

        while (tc.getHasMore()) {
            if (carrierCollector.isPresent(crr.getId())) {
                result = carrierCollector.getCarrierRequestResponse(crr.getId()).getStatus();
                tc.setHasMore(false);
                timer.cancel();
            }
        }

        return result;
    }

    private Runnable reciever() {
        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    DatagramChannel channel = DatagramChannel.open();
                    DatagramSocket socket = channel.socket();
                    InetAddress addr = InetAddress.getByName(myHost);
                    SocketAddress address = new InetSocketAddress(addr, myPort);
                    socket.bind(address);

                    ByteBuffer buffer = ByteBuffer.allocateDirect(65536);
                    while (true) {
                        SocketAddress client = channel.receive(buffer);
                        buffer.flip();

                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes, 0, bytes.length);
                        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                        GZIPInputStream gz = new GZIPInputStream(bais);
                        ObjectInputStream in = new ObjectInputStream(gz);
                        Object object = in.readObject();
                        logger.log(Level.INFO, "Connector server reciever get object ({0})", object);

                        if (object instanceof DataChunk) {
                            dataChunkCollector.addChunk((DataChunk) object);
                        }
                        buffer.clear();
                    }

                } catch (Exception e) {
                    logger.log(Level.SEVERE, myHost, e);
                }
            }
        };

        return r;
    }

    public void send(Object obj) {
        Executors.newSingleThreadExecutor().execute(sender(obj));
    }

    private Runnable sender(final Object object) {

        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    DatagramChannel channel = DatagramChannel.open();
                    channel.configureBlocking(false);
                    channel.connect(new InetSocketAddress(remoteHost, remotePort));
                    Selector selector = Selector.open();
                    channel.register(selector, SelectionKey.OP_WRITE);

                    boolean hasMore = true;
                    while (hasMore) {
                        selector.select(10000);
                        Set readyKeys = selector.selectedKeys();
                        if (readyKeys.isEmpty()) {
                            break;

                        } else {
                            Iterator iterator = readyKeys.iterator();
                            while (iterator.hasNext()) {
                                SelectionKey key = (SelectionKey) iterator.next();
                                iterator.remove();

                                if (key.isWritable()) {
                                    ArrayList<DataChunk> dataChunkList = new DataChunkList().getDataChunkList(object);
                                    for (DataChunk dataChunk : dataChunkList) {
                                        ByteArrayOutputStream baos = new ByteArrayOutputStream();
                                        GZIPOutputStream gz = new GZIPOutputStream(baos);
                                        ObjectOutputStream oout = new ObjectOutputStream(gz);
                                        oout.writeObject(dataChunk);
                                        oout.flush();
                                        oout.close();
                                        gz.flush();
                                        gz.close();

                                        channel.write(ByteBuffer.wrap(baos.toByteArray()));
                                    }
                                    key.interestOps(SelectionKey.OP_READ);
                                }
                                hasMore = false;
                            }
                        }
                    }

                    selector.close();
                    channel.close();

                } catch (Exception e) {
                    logger.log(Level.SEVERE, remoteHost, e);
                }
            }
        };

        return r;
    }

    public static ConnectorServer getInstance() {
        return ServerHolder.INSTANCE;
    }

    private static class ServerHolder {

        private static final ConnectorServer INSTANCE = new ConnectorServer();
    }
}
